/*
    消息管理相关类的实现: 队列消息管理类、消息管理类

    1. 队列消息管理类: 以队列为单位对消息数据的管理以及持久化（一个对象管理一个队列消息）, 持久化文件以消息队列名命名, 以.mqd为后缀
        属性: 文件路径(消息持久化文件的所在路径)、队列名、待推送链表、持久化映射表、待确认映射表、持久化总消息数量、持久化总有效消息数量
        方法: 插入消息、删除消息(应答消息时使用)、获取队头消息、恢复消息
    注1: 消息的管理以队列为单位进行管理和持久化，目的是为了方便垃圾回收、减少锁的冲突概率、降低垃圾回收的频率、提高容错
        如果都存放到一个大文件，那么文件中需要保存更多的消息信息才能正确恢复，同时其他影响因素也会变大
    注2: 使用链表作为队列原因: 首先队列也方便头删和尾插, 其次相比queue支持遍历
        使用持久化映射表的原因: 垃圾回收时, 会形成一个新的持久化消息文件, 则消息的偏移量也会发生改变，所以也需要对内存中持久化消息的描述进行修改
            为了能够快速查找到具体的消息, 所以采用unordered_map
        使用确认映射表的原因：当对消息进行确认时，需要根据消息id快速定位到消息，所以使用unordered_map
    注3: 持久化总消息数量和持久化总有效消息数量目的是为了判断垃圾回收时的时机。
    注4：持久化总消息数量描述的是持久化文件中所有消息数量，包括有效消息和无效消息，当插入消息时或者垃圾回收时才会变化
        持久化总有效消息数量描述的是持久化文件中有效消息的数量，当插入消息时、垃圾回收时以及应答消息时候才会变化
    注5: 消息持久化的意义是，只要消息没有被推送，就会被保存
    2. 消息管理类: 对所有队列消息数据的管理, 从而达到对所有消息的管理
        属性: 文件路径(消息持久化文件的所在路径)、管理所有队列消息对象的映射表<队列名, 队列消息对象>
        方法: 向指定队列插入消息、向指定队列应答消息、从指定队列获取队头消息、初始化队列消息(消息的恢复)、销毁指定队列消息
    注1: 消息的数据流，插入消息时，消息被插入到待推送队列，如果消息需要持久化，则也会被放入到持久化映射表和持久化文件中
        当获取队头消息之后，消息会送待推送队列删除，然后放入到待确认映射表中
        当消息被确认之后，消息会从待确认映射表中删除，同时也会从持久化映射表和持久化文件中删除(持久化文件是伪删除)
    注: 这种管理思想采用文件系统的思想
*/
#pragma once
#include "../common/message.pb.h"
#include "../common/helper.hpp"
#include "../common/logger.hpp"
#include <list>

namespace mq
{
    using MessagePtr = std::shared_ptr<Message>;
    // 队列消息类
    class QueueMessage
    {
    public:
        const int MAX_MSG_CNT = 1000;
        const int MAX_VALID_MSG_MUL = 2;
        #define FILE_SUFFIX ".mqd" // 持久化文件后缀
        using QueueMessagePtr = std::shared_ptr<QueueMessage>;

    public:
        QueueMessage(const std::string &dir, const std::string &msg_queue_name)
        {
            // 1. 构建持久化文件路径
            _file = dir + "/" + msg_queue_name + FILE_SUFFIX;
            // 2. 创建目录
            if(!FileHelper::createDir(dir))
            {
                ERR_LOG("create message dir %s failed\n", dir.c_str());
                abort();
            }

            // 3. 创建文件
            if(!mq::FileHelper(_file).exists())
            {
                createMsgFile();
                _total_cnt = _valid_cnt = 0;
            }
            else
            {
                // 4. 恢复消息
                _push_list = recoveryAllMessages(); // 恢复待推送链表
                for(auto& msg_ptr : _push_list) // 恢复持久化映射表
                {
                    _durable_map[msg_ptr->payload().properties().id()] = msg_ptr;
                }
                _total_cnt = _valid_cnt = _push_list.size();
            }
        }

    public:
        // 插入消息
        void insertMsg(const BasicProperties *properties, const std::string &body, bool is_durable_msg_queue)
        {
            // 1. 构建消息对象
            MessagePtr msg_ptr = std::make_shared<Message>();
            Message_Payload* msg_payload_ptr = msg_ptr->mutable_payload();
            msg_payload_ptr->set_body(body);
            BasicProperties* basic_properties_ptr = msg_payload_ptr->mutable_properties();
            if(properties)
            {
                DeliveryMode delivery_mode = is_durable_msg_queue ? properties->delivery_mode() : DeliveryMode::UNDURABLE;
                basic_properties_ptr->set_id(properties->id());
                basic_properties_ptr->set_delivery_mode(delivery_mode);
                basic_properties_ptr->set_routing_key(properties->routing_key());
            }
            else
            {
                DeliveryMode delivery_mode = is_durable_msg_queue ? DeliveryMode::DURABLE : DeliveryMode::UNDURABLE;
                basic_properties_ptr->set_id(UUIDHelper::uuid());
                basic_properties_ptr->set_delivery_mode(delivery_mode);
                basic_properties_ptr->set_routing_key("");
            }
            
            std::unique_lock<std::mutex> lock(_mutex);
            // 2. 添加到待推送队列
            _push_list.push_back(msg_ptr);

            if(basic_properties_ptr->delivery_mode() == DeliveryMode::DURABLE)
            {
                msg_payload_ptr->set_vaild("1"); // 只有持久化时, _valid字段才有意义
                // 3. 添加到持久化映射表
                _durable_map[basic_properties_ptr->id()] = msg_ptr;

                // 4. 写入到文件进行持久化
                insertMsgToFile(msg_ptr);

                // 5. 更新计数器
                _total_cnt++;
                _valid_cnt++;

                // 6. 检查是否需要垃圾回收
                checkGarbageCollection();
            }
        }

        // 删除消息(接收到消息应答后调用)
        void removeMsg(const std::string &id)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            // 1. 从待确认映射表中查找
            auto it = _waitting_ack_map.find(id);
            if(it == _waitting_ack_map.end()) return;

            if(it->second->payload().properties().delivery_mode() == DeliveryMode::DURABLE)
            {
                // 2. 从持久化映射表中删除
                _durable_map.erase(id);

                // 3. 从持久化文件中删除
                deleteMsgFromFile(it->second);

                // 4. 更新计数器
                _valid_cnt--;

                // 5. 检查是否需要垃圾回收
                checkGarbageCollection();
            }

            // 6. 从待确认映射表中删除
            _waitting_ack_map.erase(it); // 注意：等待映射表中的删除要在持久化删除之后，不然先删除映射表，那么Message元素直接就比删除了并且迭代器也会失效，使用的话是非法行为
        }

        // 出队头消息
        MessagePtr popFront()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            if(_push_list.empty()) return nullptr;
            // 1. 获取待推送链表头节点并pop掉
            MessagePtr msg_ptr = _push_list.front();
            _push_list.pop_front();
            // DBG_LOG("pop msg %s\n", msg_ptr->payload().properties().id().c_str());

            // 2. 添加头结点到待确认映射表, 等待应答之后再删除
            _waitting_ack_map[msg_ptr->payload().properties().id()] = msg_ptr;
            return msg_ptr;
        }

    private:
        // 判断是否需要垃圾回收(消息总数量超过MAX_MSG_CNT并且有效消息低于总消息的一半)
        void checkGarbageCollection()
        {
            if(_total_cnt > MAX_MSG_CNT && _valid_cnt * MAX_VALID_MSG_MUL < _total_cnt)
            {
                // 1. 垃圾回收
                std::list<MessagePtr> durable_list = garbageCollection();

                // 2. 更新持久化映射表
                _durable_map.clear();
                for(auto& msg_ptr : durable_list)
                {
                    _durable_map[msg_ptr->payload().properties().id()] = std::move(msg_ptr);
                }

                // 3. 更新计数器 
                _total_cnt = _valid_cnt = _durable_map.size();
            }
        }

    public:
        /*
            下面是测试使用的接口
        */
        // 获取持久化中所有消息数量
        size_t getTotalCnt()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            return _total_cnt;
        }

        // 获取持久化中有效消息数量
        size_t getValidCnt()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            return _valid_cnt;
        }

        // 获取待推送消息数量
        size_t getPushCnt()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            return _push_list.size();
        }

        // 获取等待确认消息数量
        size_t getWaittingAckCnt()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            return _waitting_ack_map.size();
        }

        // 清空
        void clear()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _push_list.clear();
            _durable_map.clear();
            _waitting_ack_map.clear();
            _total_cnt = _valid_cnt = 0;
            removeMsgFile();
        }

    private:
        /*
            下面是和持久化相关的操作
        */
        // 创建消息文件
        void createMsgFile()
        {
            if(!FileHelper::createFile(_file))
            {
                ERR_LOG("create message file %s failed\n", _file.c_str());
                abort();
            }
        }
        
        // 删除消息文件
        void removeMsgFile()
        {
            if(!FileHelper::removeFile(_file))
            {
                ERR_LOG("remove message file %s failed\n", _file.c_str());
                abort();
            }
        }

        // 插入消息到文件
        void insertMsgToFile(const MessagePtr &msg_ptr)
        {
            insertMsgHandler(_file, msg_ptr);
        }

        // 删除消息从文件(伪删除即覆盖)
        void deleteMsgFromFile(const MessagePtr& msg_ptr)
        {
            // 1. 更新消息描述
            Message_Payload* msg_payload_ptr = msg_ptr->mutable_payload();
            msg_payload_ptr->set_vaild("0");

            // 2. 消息序列化
            std::string msg_buf = msg_payload_ptr->SerializeAsString();

            // 3. 覆盖文件
            if(msg_buf.size() != msg_ptr->length())
            {
                ERR_LOG("message length error, message id %s\n", msg_ptr->payload().properties().id().c_str());
                return;
            }
            FileHelper write_file_helper(_file);
            if(!write_file_helper.write(&msg_buf[0], msg_ptr->offset(), msg_buf.size()))
                ERR_LOG("message delete failed, message id %s\n", msg_ptr->payload().properties().id().c_str());
        }

        // 恢复消息
        std::list<MessagePtr> recoveryAllMessages()
        {
            std::list<MessagePtr> msg_list;
            FileHelper read_file_helper(_file);
            size_t file_size = read_file_helper.size();
            size_t offset = 0;
            while(offset < file_size)
            {
                // 1. 读取size_t大小的payload长度描述
                size_t length = 0;
                if(!read_file_helper.read((char*)&length, offset, sizeof(size_t)))
                {
                    ERR_LOG("message length read failed, size is %zu\n", length);
                    return {};
                }

                // 2. 读取payload
                std::string payload_str(length, '\0');
                if(!read_file_helper.read(&payload_str[0], offset + sizeof(size_t), length))
                {
                    ERR_LOG("message paylod read failed\n");
                    return {};
                }
                
                // 3. 反序列化payload并构建消息对象
                MessagePtr msg_ptr = std::make_shared<Message>();
                Message_Payload* payload_ptr = msg_ptr->mutable_payload();
                if(payload_ptr->vaild() == "0") continue; // 无效消息, 直接丢弃
                payload_ptr->ParseFromString(payload_str);
                msg_ptr->set_length(payload_str.size());
                msg_ptr->set_offset(offset + sizeof(size_t));

                // 4. 插入到链表
                msg_list.push_back(msg_ptr);

                // 5. 更新offset
                offset += sizeof(size_t) + length;
            }
            return msg_list;
        }

        // 垃圾回收
        std::list<MessagePtr> garbageCollection()
        {
            // 1. 从持久化文件中读取数据
            std::list<MessagePtr> msg_list = recoveryAllMessages();

            // 2. 构建临时文件
            std::string tmp_file = _file.substr(0, _file.rfind(FILE_SUFFIX)) + ".tmp";
            FileHelper write_file_helper(tmp_file);

            // 3. 写入临时文件
            for(auto& msg_ptr : msg_list)
            {
                insertMsgHandler(tmp_file, msg_ptr);
            }

            // 4. 删除旧文件
            FileHelper::removeFile(_file);

            // 5. 更新临时文件名为源文件名
            write_file_helper.rename(tmp_file);
            return msg_list;
        }

    private:
        // 插入消息核心处理函数
        void insertMsgHandler(const std::string& file, const MessagePtr &msg_ptr)
        {
            // 1. 消息中的payload序列化
            Message_Payload* msg_payload_ptr = msg_ptr->mutable_payload();
            std::string msg_buf = msg_payload_ptr->SerializeAsString();

            // 2. 写入文件
            // 先写入size_t大小的payload长度描述, 再写入payload
            size_t length = msg_buf.size();
            FileHelper write_file_helper(file);
            size_t offset = write_file_helper.size();
            if(!write_file_helper.write((char*)&length, offset, sizeof(size_t)))
            {
                ERR_LOG("message length insert failed, message id %s\n", msg_ptr->payload().properties().id().c_str());
                if(!write_file_helper.truncate(offset)) // 如果写入失败，截断文件
                {
                    ERR_LOG("message length of %s file truncate failed\n", _file.c_str());
                    abort();
                }
                return;
            }
            size_t size = 0;
            FileHelper r(file);
            r.read((char*)&size, 0, sizeof(size_t));

            if(!write_file_helper.write(&msg_buf[0], offset + sizeof(size_t), msg_buf.size()))
            {
                ERR_LOG("message insert failed, message id %s\n", msg_ptr->payload().properties().id().c_str());
                if(!write_file_helper.truncate(offset)) // 如果写入失败，截断文件
                {
                    ERR_LOG("message of %s file truncate failed\n", _file.c_str());
                    abort();
                }
                return;
            }

            // 3. 更新消息描述
            msg_ptr->set_offset(offset + sizeof(size_t));
            msg_ptr->set_length(msg_buf.size());
            return;
        }

    private:
        std::string _file;                                         // 消息持久化文件
        std::list<MessagePtr> _push_list;                          // 待推送链表
        std::unordered_map<std::string, MessagePtr> _durable_map;  // 持久化映射表<消息id, 消息对象>
        std::unordered_map<std::string, MessagePtr> _waitting_ack_map; // 待确认映射表<消息id, 消息对象>
        size_t _total_cnt;                                         // 持久化总消息数量
        size_t _valid_cnt;                                         // 持久化中有效消息数量
        std::mutex _mutex;
    };

    // 消息管理类
    class MessageManager {
    public:
        using MessageManagerPtr = std::shared_ptr<MessageManager>;

    public:
        MessageManager() {}
        MessageManager(const std::string& dir): _dir(dir)
        {
        }

        /*
            下面采用向外提供接口的方式来从持久化文件中恢复队列消息，并没有直接在构造函数中恢复
            因为如果直接在构造函数中进行恢复，那么就需要提供数据库中所有持久化的队列名
            为了让两个模块之间解耦，然后在外部交换机模块进行融合处理
        */
        // 初始化指定队列消息
        void initQueueMessage(const std::string& queue_name)
        {
            // 1. 构建队列消息对象
            QueueMessage::QueueMessagePtr queue_msg_ptr = std::make_shared<QueueMessage>(_dir, queue_name);

            // 2. 进行管理
            std::unique_lock<std::mutex> lock(_mutex);
            _queue_msgs[queue_name] = queue_msg_ptr;
        }

        // 初始化所有队列消息
        void initQueueMessages(const std::vector<std::string>& queue_names)
        {
            for(auto& queue_name : queue_names)
            {
                initQueueMessage(queue_name);
            }
        }

        // 销毁指定队列消息
        void destroyQueueMessage(const std::string& queue_name)
        {
            QueueMessage::QueueMessagePtr queue_msg_ptr;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(queue_name);
                if(it == _queue_msgs.end()) return;
                queue_msg_ptr = it->second;
                _queue_msgs.erase(it);
            }
            queue_msg_ptr->clear();
        }
        
        // 插入消息(向指定队列插入消息)
        void insertMsg(const std::string& queue_name, const BasicProperties* properties, const std::string& body, bool is_durable_queue_msg)
        {
            QueueMessage::QueueMessagePtr queue_msg_ptr;
            {
                // 1. 获取队列消息对象
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(queue_name);
                if(it == _queue_msgs.end())
                {
                    queue_msg_ptr = std::make_shared<QueueMessage>(_dir, queue_name);
                    _queue_msgs[queue_name] = queue_msg_ptr;
                }
                else
                {
                    queue_msg_ptr = it->second;
                }
            }

            // 2. 插入消息
            queue_msg_ptr->insertMsg(properties, body, is_durable_queue_msg);
        }

        // 获取队头消息(获取指定队头消息)
        MessagePtr popFront(const std::string& queue_name)
        {
            QueueMessage::QueueMessagePtr queue_msg_ptr;
            {
                // 1. 获取队列消息对象
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(queue_name);
                if(it == _queue_msgs.end())
                    return nullptr;
                queue_msg_ptr = it->second;
            }
            // 2. 返回队头元素
            return queue_msg_ptr->popFront();
        }

        // 应答消息(应答指定队列的某个消息)
        void ackMsg(const std::string& queue_name, const std::string& msg_id)
        {
            QueueMessage::QueueMessagePtr queue_msg_ptr;
            {
                // 1. 获取队列消息对象
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(queue_name);
                if(it == _queue_msgs.end())
                {
                    DBG_LOG("queue message handler %s not exist\n", queue_name.c_str());
                    return;
                }
                queue_msg_ptr = it->second;
            }
            // 2. 删除待确认消息
            queue_msg_ptr->removeMsg(msg_id);
        }

    public:
        /*
            下面是测试相关的代码
        */
        // 获取指定队列中持久化的总消息数量
        size_t getTotalCnt(const std::string& queue_name)
        {
            QueueMessage::QueueMessagePtr queue_msg_ptr;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(queue_name);
                if(it == _queue_msgs.end())
                {
                    DBG_LOG("queue message handler %s not exist\n", queue_name.c_str());
                    return 0;
                }
                queue_msg_ptr = it->second;
            }
            return queue_msg_ptr->getTotalCnt();
        }

        // 获取指定队列中持久化的有效消息数量
        size_t getValidCnt(const std::string& queue_name)
        {
            QueueMessage::QueueMessagePtr queue_msg_ptr;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(queue_name);
                if(it == _queue_msgs.end())
                {
                    DBG_LOG("queue message handler %s not exist\n", queue_name.c_str());
                    return 0;
                }
                queue_msg_ptr = it->second;
            }
            return queue_msg_ptr->getValidCnt();
        }

        // 获指定队列中待推送消息数量
        size_t getPushCnt(const std::string& queue_name)
        {
            QueueMessage::QueueMessagePtr queue_msg_ptr;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(queue_name);
                if(it == _queue_msgs.end())
                {
                    DBG_LOG("queue message handler %s not exist\n", queue_name.c_str());
                    return 0;
                }
                queue_msg_ptr = it->second;
            }
            return queue_msg_ptr->getPushCnt();
        }

        // 获取指定队列中待确认消息数量
        size_t getWaittingAckCnt(const std::string& queue_name)
        {
            QueueMessage::QueueMessagePtr queue_msg_ptr;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(queue_name);
                if(it == _queue_msgs.end())
                {
                    DBG_LOG("queue message handler %s not exist\n", queue_name.c_str());
                    return 0;
                }
                queue_msg_ptr = it->second;
            }
            return queue_msg_ptr->getWaittingAckCnt();
        }

        // 获取队列消息数量
        size_t queueMsgSize()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            return _queue_msgs.size();
        }

        // 清空所有消息
        void clear()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            for(auto& it : _queue_msgs)
            {
                it.second->clear();
            }
        }

    private:
        std::unordered_map<std::string, QueueMessage::QueueMessagePtr> _queue_msgs; // 队列消息映射表<队列名称, 队列消息对象>
        std::string _dir; // 消息持久化文件存放目录
        std::mutex _mutex;
    };
}